# coding=utf-8
from bigflow import base
from bigflow import input
from bigflow import output
from bigflow import base, schema, transforms
from bigflow.transforms import *
from bil.load import wise_join_word_file
from bil.region.bgfl import point_join_region

"""
作用：日期在[a,b)之间，北京市的搜索"养老院"(各区的)搜索量
路径：${out_root}/cal/xxx
文件格式：
"""
out_root = "afs://wuge.afs.baidu.com:9902/user/bil-plat/users/v_libin09/***"
a = 2021042408
b = 2021042410


def create_pipeline():
    udw_conf2 = {"auth_type": "baas_identity_code", \
                 "baas_identity_code": "9U+h7CXfgnWe+wCGDgJP", \
                 "baas_user": "v_libin09", \
                 "baas_group": "g_bil"}
    job_conf = {'mapred.job.map.capacity': '5000', \
                'mapred.job.reduce.capacity': '5000', \
                'mapred.map.tasks': '5000', \
                'mapred.reduce.tasks': '5000', \
                'mapred.job.priority': 'VERY_HIGH', }

    tmp_file = "afs://wuge.afs.baidu.com:9902/user/bil-plat/users/v_libin09/***"
    pipeline = base.Pipeline.create("DAGMR", tmp_data_path=tmp_file, udw_conf=udw_conf2, default_concurrency=5000,
                                    hadoop_job_conf=job_conf)
    pipeline.add_directory(".", './')
    return pipeline


def wise_func(date, pipeline):
    wise_all = wise_join_word_file(pipeline, str(date), ["loc", "city", "time"], "word_dict.txt") \
        .filter(lambda x: x["city"] == "北京") \
        .filter(lambda x: x["loc"] != () and x["loc"] is not None) \
        .distinct() \
        .map(lambda x: (x["loc"], x))
    return wise_all


def get_iter_day():
    # range: 左闭右开：
    for date in range(a, b):
        pipeline = create_pipeline()
        wise_all = wise_func(date, pipeline)
        path = out_root + "/" + str(date) + "/"
        pipeline.write(wise_all, output.TextFile(path).partition(n=10))
        pipeline.run()


def cal_total():
    pipeline = create_pipeline()
    total = pipeline.parallelize([])
    for date in range(a, b):
        path = out_root + "/" + str(date)
        wise_all = pipeline.read(input.TextFile(path)).map(lambda x: eval(x))
        total = total.union(wise_all)

    point_regions = join_func(pipeline, total)
    path = out_root + "/" + "cal" + "/"
    pipeline.write(point_regions, output.TextFile(path).partition(n=1))
    pipeline.run()


def join_func(pipeline, total):
    point_regions = point_join_region(pipeline, total, max_level=3, concurrency=1000, is_shuffle=False) \
        .map(lambda x: (x["county_id"], x)) \
        .group_by_key() \
        .apply_values(transforms.count) \
        .flatten() \
        .map(change) \
        .filter(lambda x: x is not None) \
        .map(lambda x: str(x[0]) + "\t" + str(x[1]))
    return point_regions


def change(p):
    dic = {110100: "市辖区", 110101: "东城区", 110102: "西城区", 110103: "崇文区", 110104: "宣武区", 110105: "朝阳区", 110106: "丰台区",
           110107: "石景山区", 110108: "海淀区", 110109: "门头沟区",
           110111: "房山区", 110112: "通州区", 110113: "顺义区", 110114: "昌平区", 110115: "大兴区", 110116: "怀柔区", 110117: "平谷区"}
    if p[0] in dic:
        return [p[0], p[1]]


if __name__ == '__main__':
    get_iter_day()
    cal_total()
